package com.cnsugar.common.hbase.utils;

import com.cnsugar.common.hbase.entity.IHBaseEntity;
import com.cnsugar.common.hbase.IValueMapper;
import com.cnsugar.common.hbase.annotation.HTable;
import com.cnsugar.common.hbase.annotation.NotQualifier;
import com.cnsugar.common.hbase.annotation.Qualifier;
import com.cnsugar.common.reflect.BeanField;
import com.cnsugar.common.reflect.IFieldFilter;
import com.cnsugar.common.utils.CamelCaseUtils;
import com.cnsugar.common.reflect.ReflectUtils;
import com.cnsugar.common.utils.ProtostuffUtil;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.objenesis.Objenesis;
import org.objenesis.ObjenesisStd;

import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.text.ParseException;
import java.util.*;
import java.util.Date;

import static org.apache.hadoop.hbase.util.Bytes.SIZEOF_INT;
import static org.apache.hadoop.hbase.util.Bytes.SIZEOF_LONG;

/**
 * @Author Sugar
 * @Version 2019/1/15 10:58
 */
public final class TypeUtils {
    private static final Objenesis objenesis = new ObjenesisStd(true);
    private static final HashMap<String, String> tableMap = new HashMap<>();
    //hbase字段过滤器
    private static IFieldFilter fieldFilter = new IFieldFilter() {
        @Override
        public boolean filter(Field field) {
            return field.getAnnotation(NotQualifier.class) == null;
        }

        @Override
        public String name() {
            return "hbase";
        }
    };

    /**
     * 将Result转成java对象
     *
     * @param result
     * @param clazz
     * @param <T>
     * @return
     */
    public static <T> T resultToBean(Result result, Class<T> clazz) {
        if (result.isEmpty()) {
            return null;
        }
        T obj = objenesis.newInstance(clazz);
        List<BeanField> fields = ReflectUtils.getBeanFields(clazz, fieldFilter);
        if (obj instanceof IHBaseEntity) {//如果实现了IHBaseEntity接口
            ((IHBaseEntity) obj).setRow(Bytes.toString(result.getRow()));
        }
        List<Cell> cells = new ArrayList<>(result.listCells());
        for (BeanField field : fields) {
            if (cells.isEmpty()) {
                break;
            }
            Method method = field.getSetterMethod();
            if (method == null) {
                continue;
            }
            byte[] qualifier = getQualifier(field);
            for (int i = 0; i < cells.size(); i++) {
                Cell cell = cells.get(i);
                if (Arrays.equals(CellUtil.cloneQualifier(cell), qualifier)) {
                    setValue(obj, method, method.getParameterTypes()[0], CellUtil.cloneValue(cell));
                    cells.remove(i);
                    break;
                }
            }
        }
        return obj;
    }

    /**
     * 将Result转成java对象
     *
     * @param result
     * @param family
     * @param clazz
     * @param <T>
     * @return
     */
    public static <T> T resultToBean(Result result, String family, Class<T> clazz) {
        if (family == null) {
            return resultToBean(result, clazz);
        }
        return resultToBean(result, Bytes.toBytes(family), clazz);
    }

    /**
     * 将Result转成java对象
     *
     * @param result
     * @param family
     * @param clazz
     * @param <T>
     * @return
     */
    public static <T> T resultToBean(Result result, byte[] family, Class<T> clazz) {
        if (result.isEmpty()) {
            return null;
        }
        T obj = objenesis.newInstance(clazz);
        List<BeanField> fields = ReflectUtils.getBeanFields(clazz, fieldFilter);
        if (obj instanceof IHBaseEntity) {//如果实现了IHBaseEntity接口
            ((IHBaseEntity) obj).setRow(Bytes.toString(result.getRow()));
        }
        for (BeanField field : fields) {
            Method method = field.getSetterMethod();
            if (method == null) {
                continue;
            }
            byte[] qualifier = getQualifier(field);
            byte[] bytes = result.getValue(family, qualifier);
            if (bytes == null) {
                continue;
            }
            setValue(obj, method, method.getParameterTypes()[0], bytes);
        }
        return obj;
    }

    /**
     * 将Result转成Map<String, Object>对象
     *
     * @param result
     * @param family
     * @param valueMapper
     * @return
     */
    public static final Map<String, Object> resultToMap(Result result, String family, IValueMapper valueMapper) {
        if (family == null) {
            return resultToMap(result, valueMapper);
        }
        return resultToMap(result, Bytes.toBytes(family), valueMapper);
    }

    /**
     * 将Result转成Map<String, Object>对象
     *
     * @param result
     * @param family
     * @param valueMapper
     * @return
     */
    public static final Map<String, Object> resultToMap(Result result, byte[] family, IValueMapper valueMapper) {
        if (result.isEmpty()) {
            return null;
        }
        Map<String, Object> map = new HashMap<>();
        List<Cell> cells = result.listCells();
        cells.forEach(cell -> {
            if (!Arrays.equals(family, CellUtil.cloneFamily(cell))) {
                return;
            }
            String qualifier = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
            Object value = valueMapper.mapValue(qualifier, CellUtil.cloneValue(cell));
            map.put(qualifier, value);
        });
        return map;
    }

    /**
     * 将Result转成Map<String, Object>对象
     *
     * @param result
     * @param valueMapper
     * @return
     */
    public static final Map<String, Object> resultToMap(Result result, IValueMapper valueMapper) {
        if (result.isEmpty()) {
            return null;
        }
        Map<String, Object> map = new HashMap<>();
        List<Cell> cells = result.listCells();
        cells.forEach(cell -> {
            String column = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
            Object value = valueMapper.mapValue(column, CellUtil.cloneValue(cell));
            map.put(column, value);
        });
        return map;
    }

    /**
     * javabean转成Put对象
     *
     * @param entity
     * @param family
     * @return
     */
    public static final Put beanToPut(IHBaseEntity entity, byte[] family) {
        List<BeanField> fields = ReflectUtils.getBeanFields(entity.getClass(), fieldFilter);
        Put put = new Put(Bytes.toBytes(entity.getRow()));
        for (BeanField field : fields) {
            Method method = field.getGetterMethod();
            if (method == null) {
                continue;
            }
            byte[] qualifier = null;
            Qualifier qualifierAnnotation = field.field.getAnnotation(Qualifier.class);
            if (qualifierAnnotation != null) {
                if (qualifierAnnotation.readonly()) {//只读字段
                    continue;
                }
                qualifier = Bytes.toBytes(qualifierAnnotation.value());
            } else {
                qualifier = Bytes.toBytes(CamelCaseUtils.toUnderlineName(field.getName()));
            }
            Object value = null;
            try {
                value = method.invoke(entity);
            } catch (IllegalAccessException | InvocationTargetException e) {
                e.printStackTrace();
                continue;
            }
            if (value == null) {
                continue;
            }
            Class fieldType = method.getReturnType();
            addColumn(put, family, qualifier, fieldType, value);
        }
        return put;
    }

    /**
     * map转成Put对象
     *
     * @param map
     * @param row
     * @param family
     * @return
     */
    public static final Put mapToPut(Map<String, Object> map, String row, byte[] family) {
        Put put = new Put(Bytes.toBytes(row));
        map.entrySet().forEach(entity -> {
            if (entity.getKey() == null) {
                return;
            }
            byte[] qualifier = Bytes.toBytes(entity.getKey());
            if (entity.getValue() == null) {
                put.addColumn(family, qualifier, null);
            } else {
                Class fieldType = entity.getValue().getClass();
                addColumn(put, family, qualifier, fieldType, entity.getValue());
            }
        });
        return put;
    }

    /**
     * 获取列名
     *
     * @param field
     * @return
     */
    private static byte[] getQualifier(BeanField field) {
        Qualifier qualifier = field.field.getAnnotation(Qualifier.class);
        if (qualifier != null) {
            return Bytes.toBytes(qualifier.value());
        } else {
            return Bytes.toBytes(CamelCaseUtils.toUnderlineName(field.getName()));
        }
    }

    /**
     * 将字段值绑定到java实体类的属性
     *
     * @param obj
     * @param method
     * @param fieldType
     * @param bytes
     * @param <T>
     */
    private static <T> void setValue(T obj, Method method, Class fieldType, byte[] bytes) {
        if (bytes == null || bytes.length == 0) {
            return;
        }
        Object value = null;
        if (fieldType == String.class) {
            value = Bytes.toString(bytes);
        } else if (fieldType == Integer.class || fieldType == int.class) {
            value = Bytes.toInt(bytes);
        } else if (fieldType == Long.class || fieldType == long.class) {
            if (bytes.length == SIZEOF_INT) {
                value = Long.valueOf(Bytes.toInt(bytes));
            } else {
                value = Bytes.toLong(bytes);
            }
        } else if (fieldType == Float.class || fieldType == float.class) {
            value = Bytes.toFloat(bytes);
        } else if (fieldType == Double.class || fieldType == double.class) {
            value = Bytes.toDouble(bytes);
        } else if (fieldType == Date.class) {
            if (bytes.length == SIZEOF_LONG) {//如果日期字段在hbase中存的是时间戳（毫秒）
                value = new Date(Bytes.toLong(bytes));
            } else {
                try {
                    value = DateUtils.parseDate(Bytes.toString(bytes));
                } catch (ParseException e) {
                    e.printStackTrace();
                }
            }
        } else if (fieldType == byte[].class) {
            value = bytes;
        } else {
            value = ProtostuffUtil.deserializer(bytes, fieldType);
        }
        try {
            method.invoke(obj, value);
        } catch (InvocationTargetException | IllegalAccessException e) {
            e.printStackTrace();
        }
    }

    private static void addColumn(Put put, byte[] family, byte[] qualifier, Class fieldType, Object value) {
        if (fieldType == String.class) {
            put.addColumn(family, qualifier, Bytes.toBytes((String) value));
        } else if (fieldType == Integer.class || fieldType == int.class) {
            put.addColumn(family, qualifier, Bytes.toBytes((int) value));
        } else if (fieldType == Long.class || fieldType == long.class) {
            put.addColumn(family, qualifier, Bytes.toBytes((long) value));
        } else if (fieldType == Float.class || fieldType == float.class) {
            put.addColumn(family, qualifier, Bytes.toBytes((float) value));
        } else if (fieldType == Double.class || fieldType == double.class) {
            put.addColumn(family, qualifier, Bytes.toBytes((double) value));
        } else if (fieldType == java.util.Date.class) {
            put.addColumn(family, qualifier, Bytes.toBytes(((Date) value).getTime()));//时间用毫秒
        } else if (fieldType == byte[].class) {
            put.addColumn(family, qualifier, (byte[]) value);
        } else if (fieldType == Short.class || fieldType == short.class) {
            put.addColumn(family, qualifier, Bytes.toBytes((short) value));
        } else {
            put.addColumn(family, qualifier, Bytes.toBytes(value.toString()));
        }
    }

    /**
     * 根据类获取对应的表名
     *
     * @param clazz
     * @return
     */
    public static String getHTableName(Class<?> clazz) {
        String name = clazz.getName();
        String tableName = tableMap.get(name);
        if (tableName != null) {
            return tableName;
        }

        HTable table = clazz.getAnnotation(HTable.class);
        if (table != null) {
            if (table.value() != null && !table.value().isEmpty()) {
                tableName = table.value();
            } else {
                tableName = CamelCaseUtils.toUnderlineName(clazz.getSimpleName());
            }
        } else {
            tableName = CamelCaseUtils.toUnderlineName(clazz.getSimpleName());
        }
        tableMap.put(name, tableName);
        return tableName;
    }
}
